package com.opdar.athena.message.agent.consumers;

import com.opdar.athena.base.RedisSessionManager;
import com.opdar.athena.entities.TokenEntity;
import com.opdar.athena.message.agent.client.Client;
import com.opdar.athena.message.agent.client.ClientPool;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.util.List;

/**
 * Created by shiju on 2017/7/2.
 */
public class TransferListener implements MessageListener<String, byte[]> {
    @Autowired
    RedisSessionManager redisSessionManager;
    @Autowired
    ClientPool clientPool;
    @Autowired
    KafkaMessageListenerContainer kafkaMessageListenerContainer;
    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public void onMessage(ConsumerRecord<String, byte[]> data) {
        if(data.topic().equals("TRANSFER")){
            String targetUserId = data.key();
            String messageId = new String(data.value());
            //开始获取用户所有token和所在服务器
            List<TokenEntity> tokens = redisSessionManager.getTokenWithUserId(targetUserId);
            for(TokenEntity entity:tokens){
                if(redisSessionManager.exist(entity.getToken())){
                    if(entity.getHost()!=null && entity.getPort()!=null){
                        Client client = clientPool.getClient(entity.getHost()+":"+entity.getPort());
                        if(client != null){
                            client.transfer(messageId,entity);
                        }
                    }
                }else{
                    redisSessionManager.removeToken(entity.getToken(),entity.getUserId());
                }
            }
        }else if(data.topic().equals("MIGRATE")){
            DataInputStream dataInputStream = new DataInputStream(new ByteArrayInputStream(data.value()));
            try {
                //将destination通知给要迁移的用户
                String id = dataInputStream.readUTF();
                String source = dataInputStream.readUTF();
                String destination = dataInputStream.readUTF();
                Client sourceClient = clientPool.getClient(source);
                int type = dataInputStream.readInt();
                int status = dataInputStream.readInt();
                long num = dataInputStream.readLong();
                String token = dataInputStream.readUTF();
                sourceClient.migrate(destination,id,type,num,token);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
